package a_aa_amainpackage.lvyou_clear.lv_update

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark._

import scala.collection.mutable.ArrayBuffer

/*
*项目名: o2odata_commend
*文件名: Tuniu
*创建者: hancongcong
*创建时间:2020/1/3 11:48
*描述: TODO
*/ object Tuniu_zhengchang {
  val clust_obs = "s3a://"

  val data_base = "Tuniu"
  val collection = "tuniu_ly_detail_2207"

  val platform = "tuniu"
  val month = "7"
  val year = "2022"

  val time_stamp = "1659110400"
  val platform_Name = "tuniu_ly"


  //原始数据保存路径
  //  val source_obs_data_Path = clust_obs + s"o2o-dataproces-group/zsc/${year}/${month}18/${data_base}/${platform_Name}"
  val source_obs_data_Path = clust_obs + s"o2o-tempdata/zsc/${year}/${month}/${data_base}/${platform_Name}"
  //每个商品一个月的数据

  //价格库上月 "o2o-dataproces-group/hancongcong/tuniu/handle_price_12"
  //  val pricePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform}/${year}/handle_price_10"
  //  val pricePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform}/2021/handle_price_12"
  val pricePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/handle_price_${month.toInt - 1}"
  val newPricePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/handle_price_${month}"

  //目的地库
  val firstArrivePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/mudidi_province_${month}"
  //上月目的地库 "o2o-dataproces-group/hancongcong/tuniu/mudidi_province_12/"
  //  val lastArrivePath = clust_obs +s"o2o-dataproces-group/zsc/platform/${platform}/${year}/mudidi_province_10"
  //  val lastArrivePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform}/2021/mudidi_province_12"
  val lastArrivePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/mudidi_province_${month.toInt - 1}"
  //出发地库
  val firstdeparturePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/chufadi_province_${month}"
  //上月出发地库 firstdeparturePath lastdeparturePath "o2o-dataproces-group/hancongcong/tuniu/chufadi_province_12/"
  //  val lastdeparturePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform}/${year}/chufadi_province_10"
  //  val lastdeparturePath = clust_obs + s"o2o-dataproces-group/zsc/platform/${platform}/2021/chufadi_province_12"
  val lastdeparturePath = clust_obs + s"o2o-tempdata/zsc/platform/${platform}/${year}/chufadi_province_${month.toInt - 1}"

  def main(args: Array[String]): Unit = {

    val sc = get_sparkContext()

    val sqlContext = new SQLContext(sc)
    sqlContext.udf.register("tranTimeToString", tranTimeToString _)

    //旅游计算 销量，最后一个减第一个后平滑，价格历史取历史，最新取最新
    val ly_source_data = sqlContext.read.json(source_obs_data_Path)

    //销量价格计算
    val priceData = tuniu_travel(sqlContext.sparkSession, ly_source_data)
      .drop("salesAmount")
    println("priceData::::" + priceData.count())


    //    //对销量进行平滑
    val historySellData = hebingshuju(sc, sqlContext, priceData) //resultData:"product_Id", "sellCount"
    val historySellRdd = fangchajisuan(sqlContext, historySellData) //resultData:“product_Id”,"sell_time_list","sell_newlist", "avg_cha", "sqrt_cha"
    //当月销量与数据关联 与价格库关联
    //    val calcuData = guanlian(sqlContext, priceData, sqlContext.read.json(historySellRdd))._1
    val calcuData = guanlian(sqlContext, priceData, sqlContext.read.json(historySellRdd))
    val value = calcuData.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val product_Id = nObject.getOrDefault("product_Id", "-1").toString
      val departure = nObject.getOrDefault("departure", "-1").toString
      nObject.put("product_Id", product_Id)
      nObject.put("departure", departure)
      nObject.toJSONString
    })
    sqlContext.read.json(value).repartition(1).write.mode("overwrite").json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}_sale")
    //处理出发地和目的地 calcuData
    val resultData1 = chufadi_deal(mudidi_deal(sqlContext.read.json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}_sale"), sqlContext), sqlContext)
    resultData1.show()
    println("写入：：：：：：：：：：：：：：")
    resultData1.where("cast(salesAmount as double) >0")
      .repartition(1).write.mode("overwrite").json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}")
    //入库
    val resultData = sqlContext.read.json(s"s3a://o2o-sourcedata-${year}/obs_result_data/${year}/${month}/${platform_Name}")
    resultData.registerTempTable("resultData")
    val data: DataFrame = sqlContext.sql(
      """
        |select
        |*,
        |(sellCount*priceText) as sale
        |from
        |(select
        |*
        |from
        |resultData)
          """.stripMargin).drop("add_to_field")
    //    """.stripMargin).drop("priceText", "salesAmount", "add_to_field")
    //      .withColumnRenamed("price", "priceText")
    //      .withColumnRenamed("sale", "salesAmount")


    //cast((priceText*2) as double) as price
    // 用于618期间计算数据
    //    data.repartition(1).write.orc(clust_obs+s"o2o-dataproces-group/zyf/2021/618/${platform_Name}")
    println("途牛旅游  -" + "注意查看结果差距，不要太大！！！" + "将要使用的销售额 - 采集价格的销售额 - 价格库的销售额 ")

    data.show(false)

    //        toes(data)

    //地址再处理

    val frame = util.updaue_addressfirst.lyandyl(data, sc, sqlContext, year, month, "ly")
    val frame1 = util.jing_bianqian.updateaddress(frame, sc, sqlContext, year, month, "ly")
    val frame2 = util.update_address.cately(frame1, sc, sqlContext, month)
    frame2.dropDuplicates("product_Id").repartition(1).write.mode("overwrite").json(clust_obs + s"o2o-tempdata/shi_xinwang/tourism/${year}/${month}/${platform_Name}")
    frame2.dropDuplicates("product_Id").registerTempTable("resultData")
    sqlContext.sql(
      """
        |select count(*),sum(sellCount),sum(salesAmount),sum(salesAmount_origin),sum(salesAmount_old) from resultData
        |""".stripMargin).show(false)
    sc.stop()
  }

  //拉数
  def get_sparkConf(): SparkConf = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.29")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.mongodb.input.uri", "mongodb://root:O2Odata123!@ 192.168.0.149:27017/admin") //  mongodb://ob:O2Odata123!@ 192.168.0.56:27017/admin   args(0)
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    conf.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    conf.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    conf
  }


  def get_sparkContext(): SparkContext = {
    val conf = get_sparkConf()
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    sc
  }

  def sell_deal(ly_source_data: DataFrame, sqlContext: SQLContext): DataFrame = {
    val ly_add_data = ly_source_data.where("size(add_to_field)>1")
    ly_add_data.registerTempTable("ly_add_data")
    val sell_data = sqlContext.sql(
      """
        |select
        |*,
        |cast((add_to_field[size(add_to_field)-1].sell_count_total-add_to_field[0].sell_count_total) as bigint) as new_sell
        |from
        |ly_add_data
      """.stripMargin).dropDuplicates("product_Id").where("new_sell>0")
      .drop("sellCount").withColumnRenamed("new_sell", "sellCount")
    sell_data
  }

  def price_deal(sell_data: DataFrame, sqlContext: SQLContext): DataFrame = {
    sell_data
      .registerTempTable("sell_data")

    val price_data = sqlContext.sql(
      """
        |select
        |*,
        |case when
        |cast((split(add_to_field[0].price,'~')[0]/split(add_to_field[size(add_to_field)-1].price,'~')[0]) as double) <=1.20 and
        |cast((split(add_to_field[0].price,'~')[0]/split(add_to_field[size(add_to_field)-1].price,'~')[0]) as double) >=0.80
        |then split(add_to_field[size(add_to_field)-1].price,'~')[0]
        |else split(add_to_field[0].price,'~')[0] end as priceText
        |from
        |sell_data
      """.stripMargin).where("cast(priceText as double) > 0").dropDuplicates("product_Id")
    price_data
  }

  //销量和价格计算
  def tuniu_travel(spark: SparkSession, dataFrame: DataFrame): DataFrame = {

    val data_Value = dataFrame.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)

      //存放comment_id用作过滤相同商品
      val comment_id_arr = new JSONArray()

      val add_to_field = nObject.getJSONArray("add_to_field")
      if (null != add_to_field && add_to_field.size() >= 1) {

        val price_temp = new ArrayBuffer[Double]
        val sell_temp = new ArrayBuffer[Long]
        //最终确定价格
        var priceText = 0D

        for (i <- 0 to add_to_field.size() - 1) {
          val add_to_field_i: JSONObject = JSON.parseObject(add_to_field.get(i).toString)

          val price_add = add_to_field_i.getOrDefault("price", "-1").toString.toDouble
          if (price_add > 0) price_temp += price_add

          //          val sell_add = add_to_field_i.getOrDefault("sell_count_total", "-1").toString.toLong
          //          if (i == 0) sell_temp += sell_add
          //          if (sell_add > 0) sell_temp += sell_add

          val comment_id_add = add_to_field_i.getOrDefault("comment_id", "-1").toString
          if (!comment_id_add.equals("-1")) {
            comment_id_arr.add(comment_id_add)
          }
        }

        var comment_id = "-1"
        if (comment_id_arr.size() > 0) {
          comment_id = comment_id_arr.getString(0)
        }
        nObject.put("comment_id", comment_id)

        //价格数组中的第一个价格
        val first_Price = price_temp(0)
        //价格数组中最后一个价格
        val last_Price = price_temp.last

        //价格最终值
        if (first_Price / last_Price <= 1.20 && first_Price / last_Price >= 0.80) {
          priceText = first_Price.formatted("%.2f").toDouble
        } else {
          priceText = last_Price.formatted("%.2f").toDouble
        }


        var sellCount: Long = 0
        val addarray = add_to_field.toArray()
        for (i <- 0 to addarray.length - 2) {
          var sell0 = JSON.parseObject(addarray(i).toString).getOrDefault("sell_count_total", "0").toString.toInt
          var sell1 = JSON.parseObject(addarray(i + 1).toString).getOrDefault("sell_count_total", "0").toString.toInt
          var sellcountTemp = sell1 - sell0
          if (sellcountTemp < 0 || sellcountTemp > 1000) {
            sellcountTemp = 0
          }
          sellCount += sellcountTemp
        }

        //        val sellCount = if (sell_temp.length > 1) sell_temp.last - sell_temp(0) else 0
        nObject.put("product_title", JSON.parseObject(add_to_field.get(add_to_field.size() - 1).toString).getOrDefault("product_title", "-1"))
        nObject.put("priceText", priceText)
        nObject.put("sellCount", sellCount)
        nObject.put("salesAmount", (priceText * sellCount).formatted("%.2f").toDouble)
      }


      nObject.toString
    })

    val frame = spark.read.json(data_Value)

    val data_rdd = frame.filter("comment_id != '-1'").dropDuplicates("comment_id").toJSON.rdd
      .union(frame.filter("comment_id = '-1'").toJSON.rdd)

    val result = spark.read.json(data_rdd).drop("comment_id").dropDuplicates("product_Id")
      .where("cast(salesAmount as double)>0")
    result
  }


  def hebingshuju(sc: SparkContext, sqlContext: SQLContext, priceData: DataFrame): DataFrame = {

    /*val esData = sc.esJsonRDD("2018_tuniu_ly,2019_tuniu_ly",
      s"""
         |{
         |  "size": 0
         |  , "query": {
         |    "match_all": {}
         |  }
         |}
          """.stripMargin).values
    val esSellData = sqlContext.read.json(esData).select("product_Id", "sellCount")
      .unionAll(priceData.select("product_Id", "sellCount"))*/
    priceData
  }

  def fangchajisuan(sqlContext: SQLContext, source_sell: DataFrame): RDD[String] = {
    source_sell.registerTempTable("source_sell")
    val good_list = sqlContext.sql(
      """
        |select
        |product_Id,
        |collect_list(sellCount) over(partition by product_Id) as sell_time_list
        |from
        |source_sell
      """.stripMargin).dropDuplicates("product_Id")
    val res = good_list.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val good_list = nObject.getOrDefault("sell_time_list", "-1").toString
      var array = JSON.parseArray(good_list)
      //数组总和
      var sumi = 0.0;
      for (i <- 0 to array.size() - 1) {
        sumi += array.get(i).toString.toDouble
      }
      //数组长度
      val array_size = array.size()
      //平均值
      var avg = (sumi / array_size).toDouble
      //方差里面 (x1-x)^2 +(x2-x)^2 +......(xn-x)^2)
      var sumfangcha = 0.toDouble
      for (i <- 0 to array.size() - 1) {
        sumfangcha += (array.get(i).toString.toInt - avg) * (array.get(i).toString.toInt - avg).toDouble
      }
      //方差 方差里面/数组长度
      val fangcha_res = sumfangcha / array_size
      //标准差
      val d = Math.sqrt(fangcha_res)
      val sell_newlist = new java.util.ArrayList[Int]()
      //判断数组中是否有大于等于标准差
      for (i <- 0 to array.size() - 1) {
        if (array.get(i).toString.toDouble >= d) {
          sell_newlist.add(array.get(i).toString.toInt)
        }
      }
      //新数组放小于方差de
      val ave_list = new java.util.ArrayList[Int]()
      for (i <- 0 to array.size() - 1) {
        if (array.get(i).toString.toDouble < d) {
          ave_list.add(array.get(i).toString.toInt)
        }
      }
      //求s数组的均值
      var size_other = ave_list.size()
      var sum_other = 0.0
      for (i <- 0 to ave_list.size() - 1) {
        sum_other += ave_list.get(i).toString.toDouble
      }
      //求去掉后的均值
      var avg_other = (sum_other / size_other).toDouble

      nObject.put("sell_newlist", sell_newlist)
      nObject.put("avg_cha", avg_other)
      nObject.put("sqrt_cha", d)
      nObject.toJSONString

    })
    res
  }

  /**
    * 当月销量与数据关联 与价格库关联
    *
    * @param sqlContext
    * @param source_9   价格数据
    * @param srll_utils 销量数据
    * @return
    */
  def guanlian(sqlContext: SQLContext, source_9: DataFrame, srll_utils: DataFrame): DataFrame = {


    //销量库
    val value = srll_utils.toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      val avg_cha = nObject.getOrDefault("avg_cha", "-1").toString
      nObject.put("avg_cha", avg_cha)
      nObject.toJSONString
    })
    val srll_util = sqlContext.read.json(value).where("avg_cha != '-1' ")
    println("srll_util:")
    srll_util.registerTempTable("srll_utils")


    //对价格数据打标签
    val dataDF = source_9
      .withColumn("platformName", lit("途牛旅游"))
      .withColumn("timeStamp", lit(s"${time_stamp}"))
      .withColumn("platformName_spelling", lit(s"${platform_Name}"))
      .withColumn("platformId", lit("86"))
      .withColumn("shopType", lit("ly"))
      .withColumnRenamed("score", "shop_score") //店铺得分
    dataDF.registerTempTable("source_9")

    //列出价格数据和销量数据的  标准差
    val data1 = sqlContext.sql(
      """
        |select
        |t1.*,
        |explode(t2.sell_newlist) as sell_da,
        |t2.avg_cha
        |from
        |source_9 t1
        |join
        |srll_utils t2
        |on t1.product_Id=t2.product_Id
      """.stripMargin)

    data1.registerTempTable("data1")

    //销量大于标准差的
    val data2 = sqlContext.sql(
      """
        |select
        |*,
        |cast(avg_cha as bigint) as sell
        |from
        |data1
        |where sellCount=sell_da
      """.stripMargin).dropDuplicates("product_Id")
    data2.registerTempTable("data2")
    println("data2:")

    //低于标准差的，sells使用价格表的销量，高于标准差的使用data2里的销量
    val data3 = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when t2.product_Id is null then cast(t1.sellCount as bigint) else cast(t2.sell as bigint) end as sells
        |from
        |source_9 t1
        |left join
        |data2 t2
        |on t1.product_Id=t2.product_Id
      """.stripMargin).dropDuplicates("product_Id")
      .drop("sellCount", "salesAmount")
      .withColumnRenamed("sells", "sellCount")
    data3.registerTempTable("data3")

    println("data3::::" + data3.count())
    data3.show()

    //    val result = sqlContext.sql(
    //      s"""
    //         |select
    //         |t1.*,
    //         |
    //         |case when t2.product_Id is null then t1.priceText else t2.priceText end as price2
    //         |from
    //         |source_data t1
    //         |left join
    //         |(
    //         |select * from
    //         |price_data where priceText > 0 ) t2
    //         |on t1.product_Id=t2.product_Id
    //         |
    //       """.stripMargin).withColumnRenamed("priceText","price_origin")
    //      .withColumnRenamed("price1","priceText")
    //      .withColumnRenamed("price2","price_old")


    //与价格库关联
    val priceData = sqlContext.read.json(pricePath)
    priceData.registerTempTable("priceData")
    //计算每个商品的销售额
    val joinPriceData = sqlContext.sql(
      """
        |select
        |*,
        |(priceText*sellCount) as salesAmount_origin,
        |(price2*sellCount) as salesAmount_old,
        |(price1*sellCount) as salesAmount
        |from
        |(select
        |t1.*,
        |case when t2.product_Id is not null  and (t1.priceText/t2.priceText > 0.8 and t1.priceText/t2.priceText < 1.2) then t2.priceText else t1.priceText end as price1,
        |case when t2.product_Id is null then t1.priceText else t2.priceText end as price2
        |from
        |data3 t1
        |left join
        |priceData t2
        |on t1.product_Id=t2.product_Id)
      """.stripMargin)
      .withColumnRenamed("priceText", "priceText_origin")
      .withColumnRenamed("price1", "priceText")
      .withColumnRenamed("price2", "priceText_old")
      .dropDuplicates("product_Id")

    joinPriceData.registerTempTable("joinPriceData")
    //    joinPriceData.show()
    //生成这个月的价格库
    joinPriceData.selectExpr("product_Id", "priceText", "tranTimeToString(cast(timeStamp as String)) as timeStamp")
      .union(priceData.select("product_Id", "priceText", "timeStamp"))
      .repartition(1).write.mode("overwrite").json(newPricePath)

    //    val sum_res = sqlContext.sql(
    //      """
    //        |select
    //        |sum(sellCount) as sum_sell,
    //        |count(*) as count_good,
    //        |sum(salesAmount) as sum_sale
    //        |from
    //        |joinPriceData
    //      """.stripMargin)
    //    println("sum_res:")


    //    更新价格库,每个季度合并一次
    //    val new_price = sqlContext.sql(
    //      s"""
    //         |select
    //         |t1.product_Id,
    //         |t1.priceText,
    //         |t1.product_title product_title,'' toursim_tickets_type
    //         |from
    //         |source_data t1
    //         |left join
    //         |(
    //         |select * from
    //         |price_data where priceText > 0 ) t2
    //         |on t1.product_Id=t2.product_Id
    //         |where t2.product_Id is null
    //       """.stripMargin)
    //      .repartition(30)
    //      .write.mode(SaveMode.Overwrite).json(newPricePath)


    //    sum_res.show()
    joinPriceData
  }

  def tranTimeToString(timestamp: String): String = {
    val tm = timestamp + "000"
    val fm = new SimpleDateFormat("yyyy-MM-dd")
    val time = fm.format(new Date(tm.toLong))
    time
  }

  def mudidi_deal(calcuData: DataFrame, sqlContext: SQLContext): DataFrame = {

    val firstArriveData = calcuData.where("first_arriving not rlike ' ' and first_arriving!='0' and first_arriving!='-1'")
      .selectExpr("product_Id", "first_arriving").dropDuplicates("product_Id")
      .withColumnRenamed("product_Id", "good_id")
      .withColumnRenamed("first_arriving", "chandi")
    //当月出发地匹配结果
    val cfd_new = cfd_mudidi_pipei(firstArriveData, sqlContext)
    //        .select("good_id","province","city")
    cfd_new.registerTempTable("cfd_new")
    //和之前月份结果合并
    val cfd_last = sqlContext.read.json(lastArrivePath)
      .withColumnRenamed("product_Id", "good_id")
      .select("good_id", "province", "city")
    cfd_last.registerTempTable("cfd_last")
    //之前有的用之前的,没有的用新处理的
    val update_new = sqlContext.sql(
      """
        |select
        |t1.good_id,
        |t1.province,
        |t1.city
        |from
        |cfd_new t1
        |left join
        |cfd_last t2
        |on t1.good_id=t2.good_id
        |where t2.good_id is null
      """.stripMargin).dropDuplicates("good_id")
    //每月目的地结果
    val value = cfd_last.union(update_new)
    value.repartition(1).write.mode("overwrite").json(firstArrivePath)
    value.registerTempTable("value")

    calcuData.registerTempTable("calcuData")
    calcuData.show()
    val resultJoinMudidiData = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when t2.good_id is null then '0' else t2.province end as province_out,
        |case when t2.good_id is null then '0' else t2.city end as city_out
        |from
        |calcuData t1
        |left join
        |value t2
        |on t1.product_Id = t2.good_id
      """.stripMargin).dropDuplicates("product_Id")
    resultJoinMudidiData
  }

  def chufadi_deal(calcuData: DataFrame, sqlContext: SQLContext): DataFrame = {
    val firstArriveData = calcuData.where("departure not rlike ' ' and departure!='0' and departure!='-1'")
      .selectExpr("product_Id", "departure").dropDuplicates("product_Id")
      .withColumnRenamed("product_Id", "good_id")
      .withColumnRenamed("departure", "chandi")
    //当月出发地匹配结果
    val cfd_new = cfd_mudidi_pipei(firstArriveData, sqlContext)
    //        .select("good_id","province","city")
    cfd_new.registerTempTable("cfd_new")
    //和之前月份结果合并 firstdeparturePath lastdeparturePath
    val cfd_last = sqlContext.read.json(lastdeparturePath)
      .withColumnRenamed("product_Id", "good_id")
      .select("good_id", "province", "city")
    cfd_last.registerTempTable("cfd_last")
    //之前有的用之前的,没有的用新处理的
    val update_new = sqlContext.sql(
      """
        |select
        |t1.good_id,
        |t1.province,
        |t1.city
        |from
        |cfd_new t1
        |left join
        |cfd_last t2
        |on t1.good_id=t2.good_id
        |where t2.good_id is null
      """.stripMargin).dropDuplicates("good_id")
    //每月目的地结果
    val value = cfd_last.union(update_new)
    value.repartition(1).write.mode("overwrite").json(firstdeparturePath)
    value.registerTempTable("value")

    calcuData.registerTempTable("calcuData")
    calcuData.show()
    val resultJoinMudidiData = sqlContext.sql(
      """
        |select
        |t1.*,
        |case when t2.good_id is null then '0' else t2.province end as province_info,
        |case when t2.good_id is null then '0' else t2.city end as city_info
        |from
        |calcuData t1
        |left join
        |value t2
        |on t1.product_Id = t2.good_id
      """.stripMargin).dropDuplicates("product_Id")
    resultJoinMudidiData
  }


  def cfd_mudidi_pipei(calcuData: DataFrame, sqlContext: SQLContext): DataFrame = {
    calcuData.registerTempTable("calcuData")
    val data = sqlContext.sql(
      """
        |select
        |*,
        |case
        |when chandi rlike '咸阳'                or chandi rlike       '咸阳'                                     then  "咸阳市"
        |when chandi rlike '平凉'                or chandi rlike       '平凉'                                     then  "平凉市"
        |when chandi rlike '伊犁哈萨克'          or chandi rlike       '伊犁哈萨克'                                  then  "伊犁哈萨克自治州"
        |when chandi rlike '固原'                or chandi rlike       '固原'                                     then  "固原市"
        |when chandi rlike '巴音郭楞'        or chandi rlike       '巴音郭楞'                                 then  "巴音郭楞蒙古自治州"
        |when chandi rlike '西安'                or chandi rlike       '西安'                                     then  "西安市"
        |when chandi rlike '渭南'                or chandi rlike       '渭南'                                     then  "渭南市"
        |when chandi rlike '喀什地'              or chandi rlike       '喀什地'                                    then  "喀什地区"
        |when chandi rlike '石嘴山'              or chandi rlike       '石嘴山'                                    then  "石嘴山市"
        |when chandi rlike '海东'                or chandi rlike       '海东'                                     then  "海东市"
        |when chandi rlike '吴忠'                or chandi rlike       '吴忠'                                     then  "吴忠市"
        |when chandi rlike '榆林'                or chandi rlike       '榆林'                                     then  "榆林市"
        |when chandi rlike '阿克苏地'            or chandi rlike       '阿克苏地'                                   then  "阿克苏地区"
        |when chandi rlike '武威'                or chandi rlike       '武威'                                     then  "武威市"
        |when chandi rlike '汉中'                or chandi rlike       '汉中'                                     then  "汉中市"
        |when chandi rlike '金昌'                or chandi rlike       '金昌'                                     then  "金昌市"
        |when chandi rlike '临夏回族'            or chandi rlike       '临夏回族'                                   then  "临夏回族自治州"
        |when chandi rlike '果洛藏族'            or chandi rlike       '果洛藏族'                                   then  "果洛藏族自治州"
        |when chandi rlike '哈密'                or chandi rlike       '哈密'                                     then  "哈密市"
        |when chandi rlike '白银'                or chandi rlike       '白银'                                     then  "白银市"
        |when chandi rlike '阿勒泰地'            or chandi rlike       '阿勒泰地'                                   then  "阿勒泰地区"
        |when chandi rlike '陇南'                or chandi rlike       '陇南'                                     then  "陇南市"
        |when chandi rlike '宝鸡'                or chandi rlike       '宝鸡'                                     then  "宝鸡市"
        |when chandi rlike '海西蒙古'      or chandi rlike       '海西蒙古'                                then  "海西蒙古族藏族自治州"
        |when chandi rlike '庆阳'                or chandi rlike       '庆阳'                                     then  "庆阳市"
        |when chandi rlike '安康'                or chandi rlike       '安康'                                     then  "安康市"
        |when chandi rlike '兰州'                or chandi rlike       '兰州'                                     then  "兰州市"
        |when chandi rlike '延安'                or chandi rlike       '延安'                                     then  "延安市"
        |when chandi rlike '昌吉回族'            or chandi rlike       '昌吉回族'                                   then  "昌吉回族自治州"
        |when chandi rlike '商洛'                or chandi rlike       '商洛'                                     then  "商洛市"
        |when chandi rlike '银川'                or chandi rlike       '银川'                                     then  "银川市"
        |when chandi rlike '黄南藏族'            or chandi rlike       '黄南藏族'                                   then  "黄南藏族自治州"
        |when chandi rlike '酒泉'                or chandi rlike       '酒泉'                                     then  "酒泉市"
        |when chandi rlike '张掖'                or chandi rlike       '张掖'                                     then  "张掖市"
        |when chandi rlike '定西'                or chandi rlike       '定西'                                     then  "定西市"
        |when chandi rlike '塔城地'              or chandi rlike       '塔城地'                                    then  "塔城地区"
        |when chandi rlike '吐鲁番'              or chandi rlike       '吐鲁番'                                    then  "吐鲁番市"
        |when chandi rlike '嘉峪关'              or chandi rlike       '嘉峪关'                                    then  "嘉峪关市"
        |when chandi rlike '海北藏族'            or chandi rlike       '海北藏族'                                   then  "海北藏族自治州"
        |when chandi rlike '乌鲁木齐'            or chandi rlike       '乌鲁木齐'                                   then  "乌鲁木齐市"
        |when chandi rlike '克拉玛依'            or chandi rlike       '克拉玛依'                                   then  "克拉玛依市"
        |when chandi rlike '玉树藏族'            or chandi rlike       '玉树藏族'                                   then  "玉树藏族自治州"
        |when chandi rlike '天水'                or chandi rlike       '天水'                                     then  "天水市"
        |when chandi rlike '和田地'              or chandi rlike       '和田地'                                    then  "和田地区"
        |when chandi rlike '西宁'                or chandi rlike       '西宁'                                     then  "西宁市"
        |when chandi rlike '中卫'                or chandi rlike       '中卫'                                     then  "中卫市"
        |when chandi rlike '海南藏族'            or chandi rlike       '海南藏族'                                   then  "海南藏族自治州"
        |when chandi rlike '克孜'    or chandi rlike       '克孜'                               then  "克孜勒苏柯尔克孜自治州"
        |when chandi rlike '博尔'        or chandi rlike       '博尔'                                 then  "博尔塔拉蒙古自治州"
        |when chandi rlike '铜川'                or chandi rlike       '铜川'                                     then  "铜川市"
        |when chandi rlike '甘南'            or chandi rlike       '甘南'                                   then  "甘南藏族自治州"
        |when chandi rlike '松原'                or chandi rlike       '松原'                                     then  "松原市"
        |when chandi rlike '牡丹江'              or chandi rlike       '牡丹江'                                    then  "牡丹江市"
        |when chandi rlike '盘锦'                or chandi rlike       '盘锦'                                     then  "盘锦市"
        |when chandi rlike '伊春'                or chandi rlike       '伊春'                                     then  "伊春市"
        |when chandi rlike '通化'                or chandi rlike       '通化'                                     then  "通化市"
        |when chandi rlike '哈尔滨'              or chandi rlike       '哈尔滨'                                    then  "哈尔滨市"
        |when chandi rlike '营口'                or chandi rlike       '营口'                                     then  "营口市"
        |when chandi rlike '辽源'                or chandi rlike       '辽源'                                     then  "辽源市"
        |when chandi rlike '延边'          or chandi rlike       '延边'                                  then  "延边朝鲜族自治州"
        |when chandi rlike '白山'                or chandi rlike       '白山'                                     then  "白山市"
        |when chandi rlike '黑河'                or chandi rlike       '黑河'                                     then  "黑河市"
        |when chandi rlike '大连'                or chandi rlike       '大连'                                     then  "大连市"
        |when chandi rlike '双鸭山'              or chandi rlike       '双鸭山'                                    then  "双鸭山市"
        |when chandi rlike '大庆'                or chandi rlike       '大庆'                                     then  "大庆市"
        |when chandi rlike '沈阳'                or chandi rlike       '沈阳'                                     then  "沈阳市"
        |when chandi rlike '吉林'                or chandi rlike       '吉林'                                     then  "吉林市"
        |when chandi rlike '鸡西'                or chandi rlike       '鸡西'                                     then  "鸡西市"
        |when chandi rlike '辽阳'                or chandi rlike       '辽阳'                                     then  "辽阳市"
        |when chandi rlike '大兴安岭'            or chandi rlike       '大兴安岭'                                   then  "大兴安岭地区"
        |when chandi rlike '鹤岗'                or chandi rlike       '鹤岗'                                     then  "鹤岗市"
        |when chandi rlike '齐齐哈尔'            or chandi rlike       '齐齐哈尔'                                   then  "齐齐哈尔市"
        |when chandi rlike '抚顺'                or chandi rlike       '抚顺'                                     then  "抚顺市"
        |when chandi rlike '阜新'                or chandi rlike       '阜新'                                     then  "阜新市"
        |when chandi rlike '鞍山'                or chandi rlike       '鞍山'                                     then  "鞍山市"
        |when chandi rlike '绥化'                or chandi rlike       '绥化'                                     then  "绥化市"
        |when chandi rlike '葫芦岛'              or chandi rlike       '葫芦岛'                                    then  "葫芦岛市"
        |when chandi rlike '本溪'                or chandi rlike       '本溪'                                     then  "本溪市"
        |when chandi rlike '朝阳'                or chandi rlike       '朝阳'                                     then  "朝阳市"
        |when chandi rlike '七台河'              or chandi rlike       '七台河'                                    then  "七台河市"
        |when chandi rlike '长春'                or chandi rlike       '长春'                                     then  "长春市"
        |when chandi rlike '佳木斯'              or chandi rlike       '佳木斯'                                    then  "佳木斯市"
        |when chandi rlike '四平'                or chandi rlike       '四平'                                     then  "四平市"
        |when chandi rlike '铁岭'                or chandi rlike       '铁岭'                                     then  "铁岭市"
        |when chandi rlike '丹东'                or chandi rlike       '丹东'                                     then  "丹东市"
        |when chandi rlike '白城'                or chandi rlike       '白城'                                     then  "白城市"
        |when chandi rlike '锦州'                or chandi rlike       '锦州'                                     then  "锦州市"
        |when chandi rlike '赣州'                or chandi rlike       '赣州'                                     then  "赣州市"
        |when chandi rlike '上饶'                or chandi rlike       '上饶'                                     then  "上饶市"
        |when chandi rlike '淮北'                or chandi rlike       '淮北'                                     then  "淮北市"
        |when chandi rlike '南昌'                or chandi rlike       '南昌'                                     then  "南昌市"
        |when chandi rlike '福州'                or chandi rlike       '福州'                                     then  "福州市"
        |when chandi rlike '南平'                or chandi rlike       '南平'                                     then  "南平市"
        |when chandi rlike '合肥'                or chandi rlike       '合肥'                                     then  "合肥市"
        |when chandi rlike '徐州'                or chandi rlike       '徐州'                                     then  "徐州市"
        |when chandi rlike '烟台'                or chandi rlike       '烟台'                                     then  "烟台市"
        |when chandi rlike '金华'                or chandi rlike       '金华'                                     then  "金华市"
        |when chandi rlike '衢州'                or chandi rlike       '衢州'                                     then  "衢州市"
        |when chandi rlike '滨州'                or chandi rlike       '滨州'                                     then  "滨州市"
        |when chandi rlike '泰州'                or chandi rlike       '泰州'                                     then  "泰州市"
        |when chandi rlike '芜湖'                or chandi rlike       '芜湖'                                     then  "芜湖市"
        |when chandi rlike '济宁'                or chandi rlike       '济宁'                                     then  "济宁市"
        |when chandi rlike '枣庄'                or chandi rlike       '枣庄'                                     then  "枣庄市"
        |when chandi rlike '南通'                or chandi rlike       '南通'                                     then  "南通市"
        |when chandi rlike '泉州'                or chandi rlike       '泉州'                                     then  "泉州市"
        |when chandi rlike '滁州'                or chandi rlike       '滁州'                                     then  "滁州市"
        |when chandi rlike '济南'                or chandi rlike       '济南'                                     then  "济南市"
        |when chandi rlike '连云港'              or chandi rlike       '连云港'                                    then  "连云港市"
        |when chandi rlike '潍坊'                or chandi rlike       '潍坊'                                     then  "潍坊市"
        |when chandi rlike '扬州'                or chandi rlike       '扬州'                                     then  "扬州市"
        |when chandi rlike '铜陵'                or chandi rlike       '铜陵'                                     then  "铜陵市"
        |when chandi rlike '安庆'                or chandi rlike       '安庆'                                     then  "安庆市"
        |when chandi rlike '绍兴'                or chandi rlike       '绍兴'                                     then  "绍兴市"
        |when chandi rlike '青岛'                or chandi rlike       '青岛'                                     then  "青岛市"
        |when chandi rlike '菏泽'                or chandi rlike       '菏泽'                                     then  "菏泽市"
        |when chandi rlike '南京'                or chandi rlike       '南京'                                     then  "南京市"
        |when chandi rlike '苏州'                or chandi rlike       '苏州'    or chandi rlike       '太仓'                                  then  "苏州市"
        |when chandi rlike '丽水'                or chandi rlike       '丽水'                                     then  "丽水市"
        |when chandi rlike '湖州'                or chandi rlike       '湖州'                                     then  "湖州市"
        |when chandi rlike '长宁'                or chandi rlike       '长宁'                                     then  "长宁区"
        |when chandi rlike '静安'                or chandi rlike       '静安'                                     then  "静安区"
        |when chandi rlike '三明'                or chandi rlike       '三明'                                     then  "三明市"
        |when chandi rlike '威海'                or chandi rlike       '威海'                                     then  "威海市"
        |when chandi rlike '杭州'                or chandi rlike       '杭州'                                     then  "杭州市"
        |when chandi rlike '九江'                or chandi rlike       '九江'                                     then  "九江市"
        |when chandi rlike '淄博'                or chandi rlike       '淄博'                                     then  "淄博市"
        |when chandi rlike '宁波'                or chandi rlike       '宁波'                                     then  "宁波市"
        |when chandi rlike '吉安'                or chandi rlike       '吉安'                                     then  "吉安市"
        |when chandi rlike '淮安'                or chandi rlike       '淮安'                                     then  "淮安市"
        |when chandi rlike '台州'                or chandi rlike       '台州'                                     then  "台州市"
        |when chandi rlike '东营'                or chandi rlike       '东营'                                     then  "东营市"
        |when chandi rlike '舟山'                or chandi rlike       '舟山'                                     then  "舟山市"
        |when chandi rlike '漳州'                or chandi rlike       '漳州'                                     then  "漳州市"
        |when chandi rlike '宁德'  or   chandi rlike '福安市'                       or chandi rlike       '宁德'  or   chandi rlike '福安市'                                 then  "宁德市"
        |when chandi rlike '景德镇'              or chandi rlike       '景德镇'                                    then  "景德镇市"
        |when chandi rlike '抚州'                or chandi rlike       '抚州'                                     then  "抚州市"
        |when chandi rlike '莆田'                or chandi rlike       '莆田'                                     then  "莆田市"
        |when chandi rlike '常州'                or chandi rlike       '常州'                                     then  "常州市"
        |when chandi rlike '宿迁'                or chandi rlike       '宿迁'                                     then  "宿迁市"
        |when chandi rlike '镇江'                or chandi rlike       '镇江'                                     then  "镇江市"
        |when chandi rlike '宿州'                or chandi rlike       '宿州'                                     then  "宿州市"
        |when chandi rlike '宿州'                or chandi rlike       '宿州'                                     then  "宿州市"
        |when chandi rlike '普陀'                or chandi rlike       '普陀'                                     then  "普陀区"
        |when chandi rlike '龙岩'                or chandi rlike       '龙岩'                                     then  "龙岩市"
        |when chandi rlike '聊城'                or chandi rlike       '聊城'                                     then  "聊城市"
        |when chandi rlike '无锡'                or chandi rlike       '无锡'                                     then  "无锡市"
        |when chandi rlike '六安'                or chandi rlike       '六安'                                     then  "六安市"
        |when chandi rlike '池州'                or chandi rlike       '池州'                                     then  "池州市"
        |when chandi rlike '盐城'                or chandi rlike       '盐城'                                     then  "盐城市"
        |when chandi rlike '德州'                or chandi rlike       '德州'                                     then  "德州市"
        |when chandi rlike '亳州'                or chandi rlike       '亳州'                                     then  "亳州市"
        |when chandi rlike '新余'                or chandi rlike       '新余'                                     then  "新余市"
        |when chandi rlike '虹口'                or chandi rlike       '虹口'                                     then  "虹口区"
        |when chandi rlike '萍乡'                or chandi rlike       '萍乡'                                     then  "萍乡市"
        |when chandi rlike '泰安'                or chandi rlike       '泰安'                                     then  "泰安市"
        |when chandi rlike '闵行'                or chandi rlike       '闵行'                                     then  "闵行区"
        |when chandi rlike '日照'                or chandi rlike       '日照'                                     then  "日照市"
        |when chandi rlike '淮南'                or chandi rlike       '淮南'                                     then  "淮南市"
        |when chandi rlike '宣城'                or chandi rlike       '宣城'                                     then  "宣城市"
        |when chandi rlike '崇明'                or chandi rlike       '崇明'                                     then  "崇明区"
        |when chandi rlike '厦门'                or chandi rlike       '厦门'                                     then  "厦门市"
        |when chandi rlike '温州'                or chandi rlike       '温州'                                     then  "温州市"
        |when chandi rlike '马鞍山'              or chandi rlike       '马鞍山'                                    then  "马鞍山市"
        |when chandi rlike '嘉兴'                or chandi rlike       '嘉兴'                                     then  "嘉兴市"
        |when chandi rlike '蚌埠'                or chandi rlike       '蚌埠'                                     then  "蚌埠市"
        |when chandi rlike '黄山'                or chandi rlike       '黄山'                                     then  "黄山市"
        |when chandi rlike '阜阳'                or chandi rlike       '阜阳'                                     then  "阜阳市"
        |when chandi rlike '临沂'                or chandi rlike       '临沂'                                     then  "临沂市"
        |when chandi rlike '鹰潭'                or chandi rlike       '鹰潭'                                     then  "鹰潭市"
        |when chandi rlike '宜春'                or chandi rlike       '宜春'                                     then  "宜春市"
        |when chandi rlike '松江'                or chandi rlike       '松江'                                     then  "松江区"
        |when chandi rlike '浦东新'              or chandi rlike       '浦东新'                                    then  "浦东新区"
        |when chandi rlike '黄浦'                or chandi rlike       '黄浦'                                     then  "黄浦区"
        |when chandi rlike '奉贤'                or chandi rlike       '奉贤'                                     then  "奉贤区"
        |when chandi rlike '宝山'                or chandi rlike       '宝山'                                     then  "宝山区"
        |when chandi rlike '金山'                or chandi rlike       '金山'                                     then  "金山区"
        |when chandi rlike '嘉定'                or chandi rlike       '嘉定'                                     then  "嘉定区"
        |when chandi rlike '杨浦'                or chandi rlike       '杨浦'                                     then  "杨浦区"
        |when chandi rlike '徐汇'                or chandi rlike       '徐汇'                                     then  "徐汇区"
        |when chandi rlike '上海'                or chandi rlike       '上海'                                     then  "上海市"
        |when chandi rlike '青浦'                or chandi rlike       '青浦'                                     then  "青浦区"
        |when chandi rlike '桂林'                or chandi rlike       '桂林'                                     then  "桂林市"
        |when chandi rlike '玉林'                or chandi rlike       '玉林'                                     then  "玉林市"
        |when chandi rlike '柳州'                or chandi rlike       '柳州'                                     then  "柳州市"
        |when chandi rlike '河池'                or chandi rlike       '河池'                                     then  "河池市"
        |when chandi rlike '贺州'                or chandi rlike       '贺州'                                     then  "贺州市"
        |when chandi rlike '来宾'                or chandi rlike       '来宾'                                     then  "来宾市"
        |when chandi rlike '中山'                or chandi rlike       '中山'                                     then  "中山市"
        |when chandi rlike '潮州'                or chandi rlike       '潮州'                                     then  "潮州市"
        |when chandi rlike '南宁'                or chandi rlike       '南宁'                                     then  "南宁市"
        |when chandi rlike '贵港'                or chandi rlike       '贵港'                                     then  "贵港市"
        |when chandi rlike '清远'                or chandi rlike       '清远'                                     then  "清远市"
        |when chandi rlike '湛江'                or chandi rlike       '湛江'                                     then  "湛江市"
        |when chandi rlike '梅州'                or chandi rlike       '梅州'                                     then  "梅州市"
        |when chandi rlike '钦州'                or chandi rlike       '钦州'                                     then  "钦州市"
        |when chandi rlike '北海'                or chandi rlike       '北海'                                     then  "北海市"
        |when chandi rlike '百色'                or chandi rlike       '百色'                                     then  "百色市"
        |when chandi rlike '汕尾'                or chandi rlike       '汕尾'                                     then  "汕尾市"
        |when chandi rlike '东莞'                or chandi rlike       '东莞'                                     then  "东莞市"
        |when chandi rlike '崇左'                or chandi rlike       '崇左'                                     then  "崇左市"
        |when chandi rlike '阳江'                or chandi rlike       '阳江'                                     then  "阳江市"
        |when chandi rlike '珠海'                or chandi rlike       '珠海'                                     then  "珠海市"
        |when chandi rlike '韶关'                or chandi rlike       '韶关'                                     then  "韶关市"
        |when chandi rlike '佛山'                or chandi rlike       '佛山'                                     then  "佛山市"
        |when chandi rlike '河源'                or chandi rlike       '河源'                                     then  "河源市"
        |when chandi rlike '三亚'                or chandi rlike       '三亚'                                     then  "三亚市"
        |when chandi rlike '云浮'                or chandi rlike       '云浮'                                     then  "云浮市"
        |when chandi rlike '肇庆'                or chandi rlike       '肇庆'                                     then  "肇庆市"
        |when chandi rlike '防城港'              or chandi rlike       '防城港'                                    then  "防城港市"
        |when chandi rlike '梧州'                or chandi rlike       '梧州'                                     then  "梧州市"
        |when chandi rlike '深圳'                or chandi rlike       '深圳'                                     then  "深圳市"
        |when chandi rlike '江门'                or chandi rlike       '江门'                                     then  "江门市"
        |when chandi rlike '广州'                or chandi rlike       '广州'                                     then  "广州市"
        |when chandi rlike '茂名'                or chandi rlike       '茂名'                                     then  "茂名市"
        |when chandi rlike '揭阳'                or chandi rlike       '揭阳'                                     then  "揭阳市"
        |when chandi rlike '海口'                or chandi rlike       '海口'                                     then  "海口市"
        |when chandi rlike '汕头'                or chandi rlike       '汕头'                                     then  "汕头市"
        |when chandi rlike '惠州'                or chandi rlike       '惠州'                                     then  "惠州市"
        |when chandi rlike '儋州'                or chandi rlike       '儋州'                                     then  "儋州市"
        |when chandi rlike '三沙'                or chandi rlike       '三沙'                                     then  "三沙市"
        |when chandi rlike '丽江'                or chandi rlike       '丽江'                                     then  "丽江市"
        |when chandi rlike '唐山'                or chandi rlike       '唐山'                                     then  "唐山市"
        |when chandi rlike '大理'            or chandi rlike       '大理'                                   then  "大理白族自治州"
        |when chandi rlike '红河'      or chandi rlike       '红河'                                then  "红河哈尼族彝族自治州"
        |when chandi rlike '乐山'                or chandi rlike       '乐山'                                     then  "乐山市"
        |when chandi rlike '昭通'                or chandi rlike       '昭通'                                     then  "昭通市"
        |when chandi rlike '晋中'                or chandi rlike       '晋中'                                     then  "晋中市"
        |when chandi rlike '西双版纳'        or chandi rlike       '西双版纳'                                 then  "西双版纳傣族自治州"
        |when chandi rlike '日喀则'              or chandi rlike       '日喀则'                                    then  "日喀则市"
        |when chandi rlike '迪庆'            or chandi rlike       '迪庆'                                   then  "迪庆藏族自治州"
        |when chandi rlike '安顺'                or chandi rlike       '安顺'                                     then  "安顺市"
        |when chandi rlike '昌都'                or chandi rlike       '昌都'                                     then  "昌都市"
        |when chandi rlike '邢台'                or chandi rlike       '邢台'                                     then  "邢台市"
        |when chandi rlike '长治'                or chandi rlike       '长治'                                     then  "长治市"
        |when chandi rlike '沧州'                or chandi rlike       '沧州'                                     then  "沧州市"
        |when chandi rlike '承德'                or chandi rlike       '承德'                                     then  "承德市"
        |when chandi rlike '宜宾'                or chandi rlike       '宜宾'                                     then  "宜宾市"
        |when chandi rlike '运城'                or chandi rlike       '运城'                                     then  "运城市"
        |when chandi rlike '巴中'                or chandi rlike       '巴中'                                     then  "巴中市"
        |when chandi rlike '包头'                or chandi rlike       '包头'                                     then  "包头市"
        |when chandi rlike '保定'                or chandi rlike       '保定'                                     then  "保定市"
        |when chandi rlike '石家庄'              or chandi rlike       '石家庄'                                    then  "石家庄市"
        |when chandi rlike '衡水'                or chandi rlike       '衡水'                                     then  "衡水市"
        |when chandi rlike '昆明'                or chandi rlike       '昆明'                                     then  "昆明市"
        |when chandi rlike '雅安'                or chandi rlike       '雅安'                                     then  "雅安市"
        |when chandi rlike '铜仁'                or chandi rlike       '铜仁'                                     then  "铜仁市"
        |when chandi rlike '乌兰察布'            or chandi rlike       '乌兰察布'                                   then  "乌兰察布市"
        |when chandi rlike '甘孜'            or chandi rlike       '甘孜'                                   then  "甘孜藏族自治州"
        |when chandi rlike '山南'                or chandi rlike       '山南'                                     then  "山南市"
        |when chandi rlike '临汾'                or chandi rlike       '临汾'                                     then  "临汾市"
        |when chandi rlike '保山'                or chandi rlike       '保山'                                     then  "保山市"
        |when chandi rlike '朔州'                or chandi rlike       '朔州'                                     then  "朔州市"
        |when chandi rlike '凉山'            or chandi rlike       '凉山'                                   then  "凉山彝族自治州"
        |when chandi rlike '泸州'                or chandi rlike       '泸州'                                     then  "泸州市"
        |when chandi rlike '成都'                or chandi rlike       '成都'                                     then  "成都市"
        |when chandi rlike '鄂尔多斯'            or chandi rlike       '鄂尔多斯'                                   then  "鄂尔多斯市"
        |when chandi rlike '那曲'                or chandi rlike       '那曲'                                     then  "那曲地区"
        |when chandi rlike '黔东'      or chandi rlike       '黔东'                                then  "黔东南苗族侗族自治州"
        |when chandi rlike '通辽'                or chandi rlike       '通辽'                                     then  "通辽市"
        |when chandi rlike '绵阳'                or chandi rlike       '绵阳'                                     then  "绵阳市"
        |when chandi rlike '德阳'                or chandi rlike       '德阳'                                     then  "德阳市"
        |when chandi rlike '璧山'                or chandi rlike       '璧山'                                     then  "璧山区"
        |when chandi rlike '邯郸'                or chandi rlike       '邯郸'                                     then  "邯郸市"
        |when chandi rlike '阿拉善'            or chandi rlike       '阿拉善'                                   then  "阿拉善盟"
        |when chandi rlike '东丽'                or chandi rlike       '东丽'                                     then  "东丽区"
        |when chandi rlike '太原'                or chandi rlike       '太原'                                     then  "太原市"
        |when chandi rlike '兴安盟'              or chandi rlike       '兴安盟'                                    then  "兴安盟"
        |when chandi rlike '秦皇岛'              or chandi rlike       '秦皇岛'                                    then  "秦皇岛市"
        |when chandi rlike '锡林郭勒'          or chandi rlike       '锡林郭勒'                                  then  "锡林郭勒盟"
        |when chandi rlike '呼和浩特'            or chandi rlike       '呼和浩特'                                   then  "呼和浩特市"
        |when chandi rlike '密云'                or chandi rlike       '密云'                                     then  "密云区"
        |when chandi rlike '拉萨'                or chandi rlike       '拉萨'                                     then  "拉萨市"
        |when chandi rlike '赤峰'                or chandi rlike       '赤峰'                                     then  "赤峰市"
        |when chandi rlike '阿里地'              or chandi rlike       '阿里地'                                    then  "阿里地区"
        |when chandi rlike '临沧'                or chandi rlike       '临沧'                                     then  "临沧市"
        |when chandi rlike '眉山'                or chandi rlike       '眉山'                                     then  "眉山市"
        |when chandi rlike '忻州'  or   chandi rlike '宁武县'                       or chandi rlike       '忻州'  or   chandi rlike '宁武县'                                  then  "忻州市"
        |when chandi rlike '黔南'      or chandi rlike       '黔南'                                then  "黔南布依族苗族自治州"
        |when chandi rlike '玉溪'                or chandi rlike       '玉溪'                                     then  "玉溪市"
        |when chandi rlike '沙坪坝'              or chandi rlike       '沙坪坝'                                    then  "沙坪坝区"
        |when chandi rlike '张家口'              or chandi rlike       '张家口'                                    then  "张家口市"
        |when chandi rlike '南充'                or chandi rlike       '南充'                                     then  "南充市"
        |when chandi rlike '巴彦淖尔'            or chandi rlike       '巴彦淖尔'                                   then  "巴彦淖尔市"
        |when chandi rlike '重庆'                or chandi rlike       '重庆'                                     then  "重庆市"
        |when chandi rlike '吕梁'                or chandi rlike       '吕梁'                                     then  "吕梁市"
        |when chandi rlike '黔西'    or chandi rlike       '黔西'                               then  "黔西南布依族苗族自治州"
        |when chandi rlike '阿坝'        or chandi rlike       '阿坝'                                 then  "阿坝藏族羌族自治州"
        |when chandi rlike '广元'                or chandi rlike       '广元'                                     then  "广元市"
        |when chandi rlike '呼伦贝尔'            or chandi rlike       '呼伦贝尔'                                   then  "呼伦贝尔市"
        |when chandi rlike '北碚'                or chandi rlike       '北碚'                                     then  "北碚区"
        |when chandi rlike '延庆'                or chandi rlike       '延庆'                                     then  "延庆区"
        |when chandi rlike '文山'        or chandi rlike       '文山'                                 then  "文山壮族苗族自治州"
        |when chandi rlike '涪陵'                or chandi rlike       '涪陵'                                     then  "涪陵区"
        |when chandi rlike '遵义'                or chandi rlike       '遵义'                                     then  "遵义市"
        |when chandi rlike '大同'                or chandi rlike       '大同'                                     then  "大同市"
        |when chandi rlike '资阳'                or chandi rlike       '资阳'                                     then  "资阳市"
        |when chandi rlike '昌平'                or chandi rlike       '昌平'                                     then  "昌平区"
        |when chandi rlike '普洱'   or chandi rlike '思茅区'                        or chandi rlike       '普洱'   or chandi rlike '思茅区'                                  then  "普洱市"
        |when chandi rlike '乌海'                or chandi rlike       '乌海'                                     then  "乌海市"
        |when chandi rlike '六盘水'              or chandi rlike       '六盘水'                                    then  "六盘水市"
        |when chandi rlike '蓟州'                or chandi rlike       '蓟州'                                     then  "蓟州区"
        |when chandi rlike '巫溪县'              or chandi rlike       '巫溪县'                                    then  "巫溪县"
        |when chandi rlike '阳泉'                or chandi rlike       '阳泉'                                     then  "阳泉市"
        |when chandi rlike '怒江'          or chandi rlike       '怒江'                                  then  "怒江傈僳族自治州"
        |when chandi rlike '合川'                or chandi rlike       '合川'                                     then  "合川区"
        |when chandi rlike '怀柔'                or chandi rlike       '怀柔'                                     then  "怀柔区"
        |when chandi rlike '林芝'                or chandi rlike       '林芝'                                     then  "林芝市"
        |when chandi rlike '楚雄彝族'            or chandi rlike       '楚雄彝族'                                   then  "楚雄彝族自治州"
        |when chandi rlike '遂宁'                or chandi rlike       '遂宁'                                     then  "遂宁市"
        |when chandi rlike '北京'                or chandi rlike       '北京'                                     then  "北京市"
        |when chandi rlike '渝北'                or chandi rlike       '渝北'                                     then  "渝北区"
        |when chandi rlike '垫江'              or chandi rlike       '垫江'                                    then  "垫江县"
        |when chandi rlike '曲靖'                or chandi rlike       '曲靖'                                     then  "曲靖市"
        |when chandi rlike '巴南'                or chandi rlike       '巴南'                                     then  "巴南区"
        |when chandi rlike '天津'                or chandi rlike       '天津'                                     then  "天津市"
        |when chandi rlike '顺义'                or chandi rlike       '顺义'                                     then  "顺义区"
        |when chandi rlike '江津'                or chandi rlike       '江津'                                     then  "江津区"
        |when chandi rlike '荣昌'                or chandi rlike       '荣昌'                                     then  "荣昌区"
        |when chandi rlike '平谷'                or chandi rlike       '平谷'                                     then  "平谷区"
        |when chandi rlike '大足'                or chandi rlike       '大足'                                     then  "大足区"
        |when chandi rlike '德宏'      or chandi rlike       '德宏'                                then  "德宏傣族景颇族自治州"
        |when chandi rlike '静海'                or chandi rlike       '静海'                                     then  "静海区"
        |when chandi rlike '武清'                or chandi rlike       '武清'                                     then  "武清区"
        |when chandi rlike '海淀'                or chandi rlike       '海淀'                                     then  "海淀区"
        |when chandi rlike '西青'                or chandi rlike       '西青'                                     then  "西青区"
        |when chandi rlike '北辰'                or chandi rlike       '北辰'                                     then  "北辰区"
        |when chandi rlike '武隆县'              or chandi rlike       '武隆县'                                    then  "武隆县"
        |when chandi rlike '南开'                or chandi rlike       '南开'                                     then  "南开区"
        |when chandi rlike '黔江'                or chandi rlike       '黔江'                                     then  "黔江区"
        |when chandi rlike '通州'                or chandi rlike       '通州'                                     then  "通州区"
        |when chandi rlike '大兴'                or chandi rlike       '大兴'                                     then  "大兴区"
        |when chandi rlike '南川'                or chandi rlike       '南川'                                     then  "南川区"
        |when chandi rlike '綦江'                or chandi rlike       '綦江'                                     then  "綦江区"
        |when chandi rlike '河西'                or chandi rlike       '河西'                                     then  "河西区"
        |when chandi rlike '秀山土'      or chandi rlike       '秀山土'                                then  "秀山土家族苗族自治县"
        |when chandi rlike '东城'                or chandi rlike       '东城'                                     then  "东城区"
        |when chandi rlike '廊坊'                or chandi rlike       '廊坊'                                     then  "廊坊市"
        |when chandi rlike '晋城'                or chandi rlike       '晋城'                                     then  "晋城市"
        |when chandi rlike '自贡'                or chandi rlike       '自贡'                                     then  "自贡市"
        |when chandi rlike '攀枝花'              or chandi rlike       '攀枝花'                                    then  "攀枝花市"
        |when chandi rlike '毕节'                or chandi rlike       '毕节'                                     then  "毕节市"
        |when chandi rlike '城口县'              or chandi rlike       '城口县'                                    then  "城口县"
        |when chandi rlike '潼南'                or chandi rlike       '潼南'                                     then  "潼南区"
        |when chandi rlike '达州'                or chandi rlike       '达州'                                     then  "达州市"
        |when chandi rlike '忠县'                or chandi rlike       '忠县'                                     then  "忠县"
        |when chandi rlike '九龙坡'              or chandi rlike       '九龙坡'                                    then  "九龙坡区"
        |when chandi rlike '门头沟'              or chandi rlike       '门头沟'                                    then  "门头沟区"
        |when chandi rlike '丰台'                or chandi rlike       '丰台'                                     then  "丰台区"
        |when chandi rlike '广安'                or chandi rlike       '广安'                                     then  "广安市"
        |when chandi rlike '滨海新'              or chandi rlike       '滨海新'                                    then  "滨海新区"
        |when chandi rlike '渝中'                or chandi rlike       '渝中'                                     then  "渝中区"
        |when chandi rlike '贵阳'                or chandi rlike       '贵阳'                                     then  "贵阳市"
        |when chandi rlike '房山'                or chandi rlike       '房山'                                     then  "房山区"
        |when chandi rlike '内江'                or chandi rlike       '内江'                                     then  "内江市"
        |when chandi rlike '丰都县'              or chandi rlike       '丰都县'                                    then  "丰都县"
        |when chandi rlike '津南'                or chandi rlike       '津南'                                     then  "津南区"
        |when chandi rlike '宁河'                or chandi rlike       '宁河'                                     then  "宁河区"
        |when chandi rlike '西城'                or chandi rlike       '西城'                                     then  "西城区"
        |when chandi rlike '大渡口'              or chandi rlike       '大渡口'                                    then  "大渡口区"
        |when chandi rlike '永川'                or chandi rlike       '永川'                                     then  "永川区"
        |when chandi rlike '河北'                or chandi rlike       '河北'                                     then  "河北区"
        |when chandi rlike '石柱'          or chandi rlike       '石柱'                                  then  "石柱土家族自治县"
        |when chandi rlike '彭水'      or chandi rlike       '彭水'                                then  "彭水苗族土家族自治县"
        |when chandi rlike '巫山'              or chandi rlike       '巫山县'                                    then  "巫山县"
        |when chandi rlike '河东'                or chandi rlike       '河东'                                     then  "河东区"
        |when chandi rlike '宝坻'                or chandi rlike       '宝坻'                                     then  "宝坻区"
        |when chandi rlike '万州'                or chandi rlike       '万州'                                     then  "万州区"
        |when chandi rlike '石景山'              or chandi rlike       '石景山'                                    then  "石景山区"
        |when chandi rlike '云阳'              or chandi rlike       '云阳县'                                    then  "云阳县"
        |when chandi rlike '铜梁'                or chandi rlike       '铜梁'                                     then  "铜梁区"
        |when chandi rlike '梁平'              or chandi rlike       '梁平县'                                    then  "梁平县"
        |when chandi rlike '奉节'              or chandi rlike       '奉节县'                                    then  "奉节县"
        |when chandi rlike '开州'                or chandi rlike       '开州'                                     then  "开州区"
        |when chandi rlike '南岸'                or chandi rlike       '南岸'                                     then  "南岸区"
        |when chandi rlike '酉阳'      or chandi rlike       '酉阳'                                then  "酉阳土家族苗族自治县"
        |when chandi rlike '江北'                or chandi rlike       '江北'                                     then  "江北区"
        |when chandi rlike '和平'                or chandi rlike       '和平'                                     then  "和平区"
        |when chandi rlike '红桥'                or chandi rlike       '红桥'                                     then  "红桥区"
        |when chandi rlike '朝阳'                or chandi rlike       '朝阳'                                     then  "朝阳区"
        |when chandi rlike '长寿'                or chandi rlike       '长寿'                                     then  "长寿区"
        |when chandi rlike '那曲'                or chandi rlike       '那曲'                                     then  "那曲市"
        |when chandi rlike '洛阳'                or chandi rlike       '洛阳'                                     then  "洛阳市"
        |when chandi rlike '漯河'                or chandi rlike       '漯河'                                     then  "漯河市"
        |when chandi rlike '襄阳'                or chandi rlike       '襄阳'                                     then  "襄阳市"
        |when chandi rlike '株洲'                or chandi rlike       '株洲'                                     then  "株洲市"
        |when chandi rlike '开封'                or chandi rlike       '开封'                                     then  "开封市"
        |when chandi rlike '郑州'                or chandi rlike       '郑州'                                     then  "郑州市"
        |when chandi rlike '邵阳'                or chandi rlike       '邵阳'                                     then  "邵阳市"
        |when chandi rlike '湘潭'                or chandi rlike       '湘潭'                                     then  "湘潭市"
        |when chandi rlike '平顶山'              or chandi rlike       '平顶山'                                    then  "平顶山市"
        |when chandi rlike '湘西'      or chandi rlike       '湘西'                                then  "湘西土家族苗族自治州"
        |when chandi rlike '焦作'                or chandi rlike       '焦作'                                     then  "焦作市"
        |when chandi rlike '濮阳'                or chandi rlike       '濮阳'                                     then  "濮阳市"
        |when chandi rlike '益阳'                or chandi rlike       '益阳'                                     then  "益阳市"
        |when chandi rlike '岳阳'                or chandi rlike       '岳阳'                                     then  "岳阳市"
        |when chandi rlike '新乡'                or chandi rlike       '新乡'                                     then  "新乡市"
        |when chandi rlike '十堰'                or chandi rlike       '十堰'                                     then  "十堰市"
        |when chandi rlike '永州'                or chandi rlike       '永州'                                     then  "永州市"
        |when chandi rlike '郴州'                or chandi rlike       '郴州'                                     then  "郴州市"
        |when chandi rlike '衡阳'                or chandi rlike       '衡阳'                                     then  "衡阳市"
        |when chandi rlike '孝感'                or chandi rlike       '孝感'                                     then  "孝感市"
        |when chandi rlike '许昌'                or chandi rlike       '许昌'                                     then  "许昌市"
        |when chandi rlike '恩施'      or chandi rlike       '恩施'                                then  "恩施土家族苗族自治州"
        |when chandi rlike '荆州'                or chandi rlike       '荆州'                                     then  "荆州市"
        |when chandi rlike '鹤壁'                or chandi rlike       '鹤壁'                                     then  "鹤壁市"
        |when chandi rlike '怀化'                or chandi rlike       '怀化'                                     then  "怀化市"
        |when chandi rlike '咸宁'                or chandi rlike       '咸宁'                                     then  "咸宁市"
        |when chandi rlike '武汉'                or chandi rlike       '武汉'                                     then  "武汉市"
        |when chandi rlike '荆门'                or chandi rlike       '荆门'                                     then  "荆门市"
        |when chandi rlike '长沙'                or chandi rlike       '长沙'                                     then  "长沙市"
        |when chandi rlike '周口'                or chandi rlike       '周口'                                     then  "周口市"
        |when chandi rlike '南阳'                or chandi rlike       '南阳'                                     then  "南阳市"
        |when chandi rlike '常德'                or chandi rlike       '常德'                                     then  "常德市"
        |when chandi rlike '安阳'                or chandi rlike       '安阳'                                     then  "安阳市"
        |when chandi rlike '黄石'                or chandi rlike       '黄石'                                     then  "黄石市"
        |when chandi rlike '宜昌'                or chandi rlike       '宜昌'                                     then  "宜昌市"
        |when chandi rlike '张家界'              or chandi rlike       '张家界'                                    then  "张家界市"
        |when chandi rlike '随州'                or chandi rlike       '随州'                                     then  "随州市"
        |when chandi rlike '驻马店'              or chandi rlike       '驻马店'                                    then  "驻马店市"
        |when chandi rlike '商丘'                or chandi rlike       '商丘'                                     then  "商丘市"
        |when chandi rlike '娄底'                or chandi rlike       '娄底'                                     then  "娄底市"
        |when chandi rlike '三门峡'              or chandi rlike       '三门峡'                                    then  "三门峡市"
        |when chandi rlike '黄冈'                or chandi rlike       '黄冈'                                     then  "黄冈市"
        |when chandi rlike '鄂州'                or chandi rlike       '鄂州'                                     then  "鄂州市"
        |when chandi rlike '信阳'                or chandi rlike       '信阳'                                     then  "信阳市"
        |when chandi rlike '都匀'                or chandi rlike       '都匀'                                     then  "都匀市"
        |else '0' end as city
        |from calcuData
      """.stripMargin).drop("chandi")

    data.createOrReplaceTempView("data")

    //加载标准地址表
    val address = sqlContext.read.option("header", "true")
      .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_9/")
      .where("district='0'")
    address.createOrReplaceTempView("address")

    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case
         |when t2.city is null then 0
         |else t2.province
         |end as province
         |from data t1
         |left join
         |(select distinct province,city from address where city!='0') t2
         |on t1.city=t2.city
       """.stripMargin)
    result.registerTempTable("result")
    val pipei_mudidi = sqlContext.sql(
      """
        |select
        |*,
        |case when province='0' and city != '0' then '0' else city end as city_y
        |from
        |result
      """.stripMargin).drop("city").withColumnRenamed("city_y", "city")
      .dropDuplicates("good_id")

    //    pipei_mudidi
    //      .repartition(1).write.json(firstArrivePath)
    pipei_mudidi
  }


  def toes(data: DataFrame): Unit = {
    val value = data.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      lines
    })
    value.saveToEs(s"${year}_update_${platform}_ly/${platform}_ly_${year}_${month}", Map("es.mapping.id" -> "product_Id"))
  }


}
